package com.atguigu.gmall.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.app.func.AsyncJoinFunction;
import com.atguigu.gmall.bean.TradeSkuOrderBean;
import com.atguigu.gmall.util.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yhm
 * @create 2022-09-19 14:21
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取DWD层下单详情数据
        String topicName = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_window";
        DataStreamSource<String> orderDetailStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));

        // TODO 4 过滤脏数据转换对应的结构
        // 过滤掉用户ID和sourceType不完整的数据
        SingleOutputStreamOperator<JSONObject> jsonObjStream = orderDetailStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String sourceType = jsonObject.getString("source_type");
                String userId = jsonObject.getString("user_id");
                if (userId != null && sourceType != null) {
                    out.collect(jsonObject);
                }
            }
        });

        // TODO 5 过滤掉由于left join造成的重复数据
        SingleOutputStreamOperator<JSONObject> processStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("id");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<JSONObject> jsonObjectValueState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                jsonObjectValueState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("new_json", JSONObject.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject newJson = jsonObjectValueState.value();
                if (newJson == null) {
                    // 当前id的第一条数据
                    jsonObjectValueState.update(value);
                    TimerService timerService = ctx.timerService();
                    // 使用处理时间作为定时器的触发
                    timerService.registerProcessingTimeTimer(System.currentTimeMillis() + 5000L);
                } else {
                    // 后续的数据只需要按照时间更新状态数据
                    long newRt = newJson.getLong("rt") * 1000L;
                    long rt = value.getLong("rt") * 1000L;
                    if (rt > newRt) {
                        jsonObjectValueState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                out.collect(jsonObjectValueState.value());
                jsonObjectValueState.clear();
            }
        });

        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = processStream.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject value) throws Exception {
                return TradeSkuOrderBean.builder()
                        .skuId(value.getString("sku_id"))
                        .skuName(value.getString("sku_name"))
                        .orderAmount(Double.valueOf(value.getString("split_total_amount")))
                        .originalAmount(Double.valueOf(value.getString("split_original_amount")))
                        .activityAmount(Double.valueOf(value.getString("split_activity_amount") == null ? "0" : value.getString("split_activity_amount")))
                        .couponAmount(Double.valueOf(value.getString("split_coupon_amount") == null ? "0" : value.getString("split_coupon_amount")))
                        .ts(value.getLong("rt") * 1000L)
                        .build();

            }
        });


        // TODO 6 开窗聚合统计金额
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeSkuOrderBean>() {
            @Override
            public long extractTimestamp(TradeSkuOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        })).keyBy(new KeySelector<TradeSkuOrderBean, String>() {
            @Override
            public String getKey(TradeSkuOrderBean value) throws Exception {
                return value.getSkuId();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                        value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                        value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                        return value1;
                    }
                }, new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean next = input.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        reduceStream.print("reduce>>>");

        // TODO 7 维度关联DIM层数据
        // 7.1 关联sku表
//        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoBeanStream = reduceStream.map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
//            DruidDataSource dataSource = null;
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                dataSource = DruidPhoenixDSUtil.getDataSource();
//            }
//
//            @Override
//            public TradeSkuOrderBean map(TradeSkuOrderBean value) throws Exception {
//                DruidPooledConnection connection = null;
//                try {
//                    connection = dataSource.getConnection();
//                    JSONObject dimSkuInfo = DimUtil.getDimInfo(connection, "DIM_SKU_INFO", value.getSkuId());
//                    value.setSpuId(dimSkuInfo.getString("spuId"));
//                    value.setTrademarkId(dimSkuInfo.getString("tmId"));
//                    value.setCategory3Id(dimSkuInfo.getString("category3Id"));
//                } catch (Exception e) {
//                    e.printStackTrace();
//                } finally {
//                    if (connection != null) {
//                        try {
//                            connection.close();
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                        }
//                    }
//                }
//                return value;
//            }
//        });

        // 关联sku_info
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoBeanStream = AsyncDataStream.unorderedWait(reduceStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_SKU_INFO") {
            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setSpuId(dimSkuInfo.getString("spuId"));
                input.setTrademarkId(dimSkuInfo.getString("tmId"));
                input.setCategory3Id(dimSkuInfo.getString("category3Id"));
            }

            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getSkuId();
            }
        }, 60L, TimeUnit.SECONDS);

        skuInfoBeanStream.print("skuInfo>>>");

        // 7.2 关联spu表
        SingleOutputStreamOperator<TradeSkuOrderBean> spuNameBeanStream = AsyncDataStream.unorderedWait(skuInfoBeanStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_SPU_INFO") {
            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getSpuId();
            }

            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setSpuName(dimSkuInfo.getString("spuName"));
            }
        }, 60L, TimeUnit.SECONDS);
        spuNameBeanStream.print("spu");

        // 7.3 关联品牌表
        SingleOutputStreamOperator<TradeSkuOrderBean> tmNameStream = AsyncDataStream.unorderedWait(spuNameBeanStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_BASE_TRADEMARK") {
            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getTrademarkId();
            }

            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setTrademarkName(dimSkuInfo.getString("tmName"));
            }
        }, 60L, TimeUnit.SECONDS);
        tmNameStream.print("tm>>>");

        // 7.4 关联标签表
        SingleOutputStreamOperator<TradeSkuOrderBean> category3BeanStream = AsyncDataStream.unorderedWait(tmNameStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_BASE_CATEGORY3") {
            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getCategory3Id();
            }

            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setCategory3Name(dimSkuInfo.getString("name"));
                input.setCategory2Id(dimSkuInfo.getString("category2Id"));
            }
        }, 60L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeSkuOrderBean> category2BeanStream = AsyncDataStream.unorderedWait(category3BeanStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_BASE_CATEGORY2") {
            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getCategory2Id();
            }

            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setCategory2Name(dimSkuInfo.getString("name"));
                input.setCategory1Id(dimSkuInfo.getString("category1Id"));
            }
        }, 60L, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeSkuOrderBean> category1BeanStream = AsyncDataStream.unorderedWait(category2BeanStream, new AsyncJoinFunction<TradeSkuOrderBean>("DIM_BASE_CATEGORY1") {
            @Override
            public String getKey(TradeSkuOrderBean input) {
                return input.getCategory1Id();
            }

            @Override
            public void join(TradeSkuOrderBean input, JSONObject dimSkuInfo) {
                input.setCategory1Name(dimSkuInfo.getString("name"));
            }
        }, 60L, TimeUnit.SECONDS);

        category1BeanStream.print("category1>>>");

        // TODO 8 写出到clickHouse
        category1BeanStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_trade_sku_order_window values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        // TODO 9 执行任务
        env.execute(groupId);

    }
}
